/**
 * 
 */
package ai.moran.demo.utils;

import static com.google.common.base.Charsets.UTF_8;

import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;

import io.etcd.jetcd.ByteSequence;
import io.etcd.jetcd.Client;
import io.etcd.jetcd.Watch.Watcher;
import io.etcd.jetcd.kv.GetResponse;
import io.etcd.jetcd.options.DeleteOption;
import io.etcd.jetcd.options.GetOption;
import io.etcd.jetcd.options.WatchOption;
import io.etcd.jetcd.watch.WatchEvent;

/**
 * etcd 操作工具，包括启动监听和操作etcd v3 版本协议，只测试功能，未添加log
 * 
 * @version 1.0
 * @author zhangyanhua
 * @date 2019年10月29日 下午4:30:57
 */
public class EtcdUtil
{
    // etcl客户端链接
    private static Client etcdClient = null;

    // 链接初始化
    public static synchronized Client getEtclClient()
    {
        if (etcdClient == null)
        {
            String[] urls = "http://47.95.119.33:2379".split(",");
            etcdClient = Client.builder().endpoints(urls).build();
        }
        return etcdClient;
    }

    /**
     * 新增或者修改指定的配置
     * 
     * @param key
     * @param value
     * @throws Exception
     * @author zhangyanhua
     * @date 2019年10月29日 下午4:41:06
     */
    public static void putEtcdValueByKey(String key, String value) throws Exception
    {
        Client client = EtcdUtil.getEtclClient();
        client.getKVClient().put(ByteSequence.from(key, UTF_8), ByteSequence.from(value, UTF_8)).get();
        
    }

    /**
     * 查询指定的key名称对应的value
     *
     */
    public static String getEtcdValueByKey(String key) throws Exception
    {
        Client client = EtcdUtil.getEtclClient();
        GetResponse getResponse = client.getKVClient()
                .get(ByteSequence.from(key, UTF_8), GetOption.newBuilder().build()).get();
        
        if (getResponse.getKvs().isEmpty())
        {
            return null;
        }
        return getResponse.getKvs().get(0).getValue().toString(UTF_8);
    }

    /**获取key前缀相同的*/
    public static Map<String,String> getEtcdValueByKeyPrefix(String keyPrefix) throws Exception
    {
        Client client = EtcdUtil.getEtclClient();
        GetResponse getResponse = client.getKVClient()
                .get(ByteSequence.from(keyPrefix, UTF_8), GetOption.newBuilder().withPrefix(ByteSequence.from(keyPrefix,UTF_8)).build()).get();

        Map <String,String> map=new HashMap<>(16);
        getResponse.getKvs().forEach(keyValue -> map.put(keyValue.getKey().toString(UTF_8),keyValue.getValue().toString(UTF_8)));
        return map;
    }

    /**
     * 删除指定的配置
     * 
     * @param key
     * @throws InterruptedException
     * @throws ExecutionException
     * @author zhangyanhua
     * @date 2019年10月29日 下午4:53:24
     */
    public static void deleteEtcdValueByKey(String key) throws InterruptedException, ExecutionException
    {
        Client client = EtcdUtil.getEtclClient();
        client.getKVClient().delete(ByteSequence.from(key, UTF_8)).get();
        
    }

    public static void deleteEtcdValueByKeyPrefix(String keyPrefix) throws InterruptedException, ExecutionException
    {
        Client client = EtcdUtil.getEtclClient();
        client.getKVClient().delete(ByteSequence.from(keyPrefix, UTF_8),DeleteOption.newBuilder().withPrefix(ByteSequence.from(keyPrefix,UTF_8)).build()).get();
        
    }

    /**
     * 持续监控某个key变化的方法，执行后如果key有变化会被监控到，输入结果如下
     *  watch type= "PUT", key= "zyh1", value= "zyh1-value" 
     *  watch type= "PUT", key= "zyh1", value= "zyh1-value111" 
     *  watch type= "DELETE", key= "zyh1", value= ""
     * 
     * @param key
     * @throws Exception
     * @author zhangyanhua
     * @date 2019年10月29日 下午5:26:09
     */
    public static void watchEtcdKey(String key) throws Exception
    {
        Client client = EtcdUtil.getEtclClient();
        // 最大事件数量
        Integer maxEvents = Integer.MAX_VALUE;
        CountDownLatch latch = new CountDownLatch(maxEvents);
        Watcher watcher = null;
        try
        {
            ByteSequence watchKey = ByteSequence.from(key, UTF_8);
            WatchOption watchOpts = WatchOption.newBuilder().build();

            watcher = client.getWatchClient().watch(watchKey, watchOpts, response -> {
                for (WatchEvent event : response.getEvents())
                {
                    System.out.println("watch type= \"" + event.getEventType().toString() + "\",  key= \""
                            + Optional.ofNullable(event.getKeyValue().getKey()).map(bs -> bs.toString(UTF_8)).orElse("")
                            + "\",  value= \"" + Optional.ofNullable(event.getKeyValue().getValue())
                                    .map(bs -> bs.toString(UTF_8)).orElse("")
                            + "\"");
                }

                latch.countDown();
            });

            latch.await();
        }
        catch (Exception e)
        {
            if (watcher != null)
            {
                watcher.close();
                
                etcdClient = null;
            }
            throw e;
        }
    }

    /**
     * TODO
     * 
     * @param args
     * @throws Exception
     * @author zhangyanhua
     * @date 2019年10月29日 下午6:01:54
     */
    public static void main(String[] args) throws Exception
    {
        String key = "/test/test.json";
        String value = "update";
        String key1 = "/test/test1.json";
        String value1 = "update1";
        System.out.println("**** 测试方法开始 ****");
        EtcdUtil.putEtcdValueByKey(key,value);
        EtcdUtil.putEtcdValueByKey(key1,value1);

        System.out.println(EtcdUtil.getEtcdValueByKeyPrefix("/test"));
        EtcdUtil.deleteEtcdValueByKeyPrefix("/test");
        System.out.println(EtcdUtil.getEtcdValueByKeyPrefix("/test"));

    }
}